package org.pizza.rocket.core.listener;

import org.pizza.rocket.core.annotation.RocketMessageListener;
import org.pizza.rocket.core.consumer.BatchConsumerWrapper;
import org.pizza.rocket.core.consumer.ConsumerWrapper;
import org.pizza.rocket.core.consumer.OrderConsumerWrapper;
import org.pizza.rocket.core.consumer.SimpleConsumerWrapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.StandardEnvironment;

import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author 高巍
 * email kynb_club@163.com
 * @since 2020/11/23 2:59 下午
 */
@Slf4j
public class RocketMessageListenerRegister implements ApplicationContextAware, SmartInitializingSingleton {
    private ConfigurableApplicationContext applicationContext;
    private final StandardEnvironment environment;
    private final ObjectMapper objectMapper;
    private final AtomicLong consumerCounter = new AtomicLong(0);
    private final AtomicLong orderConsumerCounter = new AtomicLong(0);
    private final AtomicLong batchConsumerCounter = new AtomicLong(0);

    public RocketMessageListenerRegister(StandardEnvironment environment, ObjectMapper objectMapper) {
        this.environment = environment;
        this.objectMapper = objectMapper;
    }

    @Override
    public void afterSingletonsInstantiated() {
        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMessageListener.class);
        beans.forEach(this::registerListener);
    }

    private void registerListener(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
        RocketMessageListener annotation = clazz.getAnnotation(RocketMessageListener.class);
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
        ConsumerWrapper consumerWrapper=null;
        String containerBeanName=null;
        //普通消息
        if (SimpleRocketMQListener.class.isAssignableFrom(bean.getClass())) {
            //注册
            containerBeanName = String.format("%s_%s", SimpleConsumerWrapper.class.getName(), consumerCounter.incrementAndGet());
            genericApplicationContext.registerBean(containerBeanName, SimpleConsumerWrapper.class,()->this.createConsumer(annotation, bean));
            //获取
            consumerWrapper = genericApplicationContext.getBean(containerBeanName, SimpleConsumerWrapper.class);
        }
        //批量消息
        if (BatchRocketMQListener.class.isAssignableFrom(bean.getClass())) {
            //注册
             containerBeanName = String.format("%s_%s", BatchConsumerWrapper.class.getName(), batchConsumerCounter.incrementAndGet());
            genericApplicationContext.registerBean(containerBeanName, BatchConsumerWrapper.class,()->this.createBatchConsumer(annotation, bean));
            //获取
             consumerWrapper = genericApplicationContext.getBean(containerBeanName, BatchConsumerWrapper.class);
        }
        //顺序消息
        if (OrderRocketMQListener.class.isAssignableFrom(bean.getClass())) {
            //注册
             containerBeanName = String.format("%s_%s", OrderConsumerWrapper.class.getName(), orderConsumerCounter.incrementAndGet());
            genericApplicationContext.registerBean(containerBeanName, OrderConsumerWrapper.class,()->this.createOrderConsumer(annotation, bean));
            //获取
             consumerWrapper = genericApplicationContext.getBean(containerBeanName, OrderConsumerWrapper.class);
        }
        if (consumerWrapper == null) {
            //没有匹配的实现
            throw new IllegalStateException(clazz + " is not instance of " + SimpleRocketMQListener.class.getName() +"||"+BatchRocketMQListener.class.getName()+"||"+OrderRocketMQListener.class.getName());
        }
        //启动
        if (!consumerWrapper.isRunning()) {
            try {
                consumerWrapper.start();
            } catch (Exception e) {
                log.error("Started consumer failed. {}", consumerWrapper, e);
                throw new RuntimeException(e);
            }
        }
        log.info("register listener success! listenerBeanName:{} ==> containerBeanName:{} ==> group:{} ==> topic:{} ==> tag:{}"
                , beanName
                , containerBeanName
                , this.environment.resolvePlaceholders(annotation.groupName())
                , this.environment.resolvePlaceholders(annotation.topic())
                , this.environment.resolvePlaceholders(annotation.tag()));
    }

    private OrderConsumerWrapper createOrderConsumer(RocketMessageListener annotation, Object bean) {
        OrderConsumerWrapper orderConsumerWrapper = new OrderConsumerWrapper();
        orderConsumerWrapper.setMqListener((OrderRocketMQListener<?>) bean);
        orderConsumerWrapper.setEnvironment(this.environment);
        orderConsumerWrapper.setAnnotation(annotation);
        orderConsumerWrapper.setObjectMapper(this.objectMapper);
        return orderConsumerWrapper;
    }

    private BatchConsumerWrapper createBatchConsumer(RocketMessageListener annotation, Object bean) {
        BatchConsumerWrapper batchConsumerWrapper = new BatchConsumerWrapper();
        batchConsumerWrapper.setMqListener((BatchRocketMQListener<?>) bean);
        batchConsumerWrapper.setEnvironment(this.environment);
        batchConsumerWrapper.setAnnotation(annotation);
        batchConsumerWrapper.setObjectMapper(this.objectMapper);
        return batchConsumerWrapper;
    }

    private SimpleConsumerWrapper createConsumer(RocketMessageListener annotation, Object bean) {
        SimpleConsumerWrapper simpleConsumerWrapper = new SimpleConsumerWrapper();
        simpleConsumerWrapper.setMqListener((SimpleRocketMQListener<?>) bean);
        simpleConsumerWrapper.setEnvironment(this.environment);
        simpleConsumerWrapper.setAnnotation(annotation);
        simpleConsumerWrapper.setObjectMapper(this.objectMapper);
        return simpleConsumerWrapper;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }
}
